package com.chatmcp.mcprouter.service.impl;

import com.chatmcp.mcprouter.config.AppConfig;
import com.chatmcp.mcprouter.config.McpServersConfig;
import com.chatmcp.mcprouter.dto.ProxyInfoDto;
import com.chatmcp.mcprouter.dto.ServerConfigDto;
import com.chatmcp.mcprouter.dto.SseClientDto;
import com.chatmcp.mcprouter.jsonrpc.JsonRpcConstants;
import com.chatmcp.mcprouter.jsonrpc.JsonRpcError;
import com.chatmcp.mcprouter.jsonrpc.JsonRpcNotification;
import com.chatmcp.mcprouter.jsonrpc.JsonRpcRequest;
import com.chatmcp.mcprouter.jsonrpc.JsonRpcResponse;
import com.chatmcp.mcprouter.mapper.ServerLogMapper;
import com.chatmcp.mcprouter.model.ServerLog;
import com.chatmcp.mcprouter.model.SseSession;
import com.chatmcp.mcprouter.service.CacheService;
import com.chatmcp.mcprouter.service.McpClientService;
import com.chatmcp.mcprouter.service.McpServerService;
import com.chatmcp.mcprouter.service.ProxyInfoService;
import com.chatmcp.mcprouter.service.ProxyService;
import com.chatmcp.mcprouter.service.SessionManager;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 代理服务实现类
 * 只包含进程管理相关功能的改进
 */
@Slf4j
@Service
public class ProxyServiceImpl implements ProxyService {

    private static final String PROXY_INFO_PREFIX = "proxy_info:";
    private static final long PROXY_INFO_TTL = 24 * 60 * 60; // 24小时
    private static final long DEFAULT_TIMEOUT = 180000; // 3分钟超时
    private static final long HEARTBEAT_INTERVAL = 30000; // 30秒发送一次心跳

    private final AppConfig appConfig;
    private final CacheService cacheService;
    private final McpServerService mcpServerService;
    private final McpClientService mcpClientService;
    private final ServerLogMapper serverLogMapper;
    private final ObjectMapper objectMapper;
    private final McpServersConfig mcpServersConfig;
    private final SessionManager sessionManager;
    private final ProxyInfoService proxyInfoService;
    
    // 客户端缓存
    private final Map<String, McpClientService> clientCache = new ConcurrentHashMap<>();
    
    // SSE客户端映射
    private final Map<String, SseClientDto> sseClientMap = new ConcurrentHashMap<>();
    
    // McpClient工厂
    private final McpClientFactory mcpClientFactory;
    
    // 任务执行器
    private final ScheduledExecutorService taskExecutor = Executors.newScheduledThreadPool(2);

    @Autowired
    public ProxyServiceImpl(ObjectMapper objectMapper, McpServerService mcpServerService, 
                           ProxyInfoService proxyInfoService, AppConfig appConfig,
                           CacheService cacheService, McpClientService mcpClientService,
                           ServerLogMapper serverLogMapper, McpServersConfig mcpServersConfig,
                           SessionManager sessionManager) {
        this.objectMapper = objectMapper;
        this.mcpServerService = mcpServerService;
        this.proxyInfoService = proxyInfoService;
        this.appConfig = appConfig;
        this.cacheService = cacheService;
        this.mcpClientService = mcpClientService;
        this.serverLogMapper = serverLogMapper;
        this.mcpServersConfig = mcpServersConfig;
        this.sessionManager = sessionManager;
        this.mcpClientFactory = new McpClientFactory();
    }
    
    /**
     * McpClient工厂类
     */
    private class McpClientFactory {
        public McpClientService createClient() {
            return new McpClientServiceImpl(objectMapper, null);
        }
    }

    // 存储客户端会话信息和SSE发射器
    private static final Map<String, Map<String, McpClientService>> clientMap = new ConcurrentHashMap<>();
    private static final Map<String, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

    // 存储心跳任务调度器
    private static final Map<String, ScheduledExecutorService> heartbeatSchedulers = new ConcurrentHashMap<>();
    
    // 存储消息队列和重试调度器
    private static final Map<String, BlockingQueue<String>> messageQueues = new ConcurrentHashMap<>();
    private static final Map<String, ScheduledExecutorService> messageProcessors = new ConcurrentHashMap<>();
    
    // 消息处理配置
    private static final int MAX_QUEUE_SIZE = 1000;
    private static final int MAX_RETRY_COUNT = 3;
    private static final long RETRY_DELAY_MS = 1000;

    // 本地缓存代理信息
    private final Map<String, ProxyInfoDto> proxyInfoCache = new ConcurrentHashMap<>();

    @Override
    public void storeProxyInfo(String sessionId, ProxyInfoDto proxyInfo) {
        if (sessionId == null || sessionId.isEmpty() || proxyInfo == null) {
            return;
        }
        
        // 存储到内存缓存
        proxyInfoCache.put(sessionId, proxyInfo);
        
        // 存储到Redis缓存
        String cacheKey = PROXY_INFO_PREFIX + sessionId;
        cacheService.set(cacheKey, proxyInfo, PROXY_INFO_TTL, TimeUnit.SECONDS);
        log.debug("Stored proxy info for session {}", sessionId);
    }

    @Override
    public ProxyInfoDto getProxyInfo(String sessionId) {
        if (sessionId == null || sessionId.isEmpty()) {
            return null;
        }
        
        // 先从内存缓存获取
        ProxyInfoDto proxyInfo = proxyInfoCache.get(sessionId);
        if (proxyInfo != null) {
            log.debug("Retrieved proxy info from memory cache for session {}", sessionId);
            return proxyInfo;
        }
        
        // 从Redis缓存获取
        String cacheKey = PROXY_INFO_PREFIX + sessionId;
        ProxyInfoDto cachedInfo = cacheService.get(cacheKey, ProxyInfoDto.class);
        if (cachedInfo != null) {
            // 存入内存缓存
            proxyInfoCache.put(sessionId, cachedInfo);
            log.debug("Retrieved proxy info from Redis cache for session {}", sessionId);
        }
        
        return cachedInfo;
    }

    @Override
    public String processSseMessage(String serverKey, String message, ProxyInfoDto proxyInfo) {
        if (proxyInfo == null || proxyInfo.getSessionId() == null) {
            log.warn("No valid session info provided for message processing");
            try {
                return createErrorResponse("0", -32602, "Invalid session information");
            } catch (Exception e) {
                log.error("Failed to create error response: {}", e.getMessage(), e);
                return "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32602,\"message\":\"Invalid session information\"},\"id\":\"0\"}";
            }
        }
        
        String sessionId = proxyInfo.getSessionId();
        log.debug("Processing SSE message for server key {} and session {}", serverKey, sessionId);
        
        // 确保消息队列存在
        if (!messageQueues.containsKey(sessionId)) {
            createMessageQueue(sessionId);
        }
        
        // 将消息添加到队列
        BlockingQueue<String> queue = messageQueues.get(sessionId);
        if (queue == null) {
            log.error("Failed to get message queue for session {}", sessionId);
            try {
                return createErrorResponse("0", -32603, "Internal error: message queue not available");
            } catch (Exception e) {
                log.error("Failed to create error response: {}", e.getMessage(), e);
                return "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Internal error\"},\"id\":\"0\"}";
            }
        }
        
        // 尝试添加消息到队列
        boolean added = queue.offer(message);
        if (!added) {
            log.error("Message queue is full for session {}", sessionId);
            try {
                return createErrorResponse("0", -32603, "Message queue is full, please try again later");
            } catch (Exception e) {
                log.error("Failed to create error response: {}", e.getMessage(), e);
                return "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Message queue is full\"},\"id\":\"0\"}";
            }
        }
        
        // 非阻塞方式返回，实际处理由消息队列异步完成
        return "{\"jsonrpc\":\"2.0\",\"result\":\"Message accepted for processing\",\"id\":\"0\"}";
    }
    
    /**
     * 实际处理SSE消息的核心逻辑
     *
     * @param serverKey 服务器键
     * @param message 消息内容
     * @param proxyInfo 代理信息
     * @return 处理结果
     * @throws Exception 如果处理失败
     */
    private String doProcessSseMessage(String serverKey, String message, ProxyInfoDto proxyInfo) throws Exception {
        // 获取服务器配置
        ServerConfigDto serverConfig;
        if (proxyInfo != null && proxyInfo.getMcpServerId() != null && !proxyInfo.getMcpServerId().isEmpty()) {
            serverConfig = mcpServerService.getServerConfig(proxyInfo.getMcpServerId());
            if (serverConfig == null) {
                log.warn("Invalid MCP server ID: {}", proxyInfo.getMcpServerId());
                return createErrorResponse("0", -32602, "Invalid MCP server ID");
            }
        } else {
            serverConfig = mcpServerService.getServerConfig(serverKey);
            if (serverConfig == null) {
                log.warn("Invalid server key: {}", serverKey);
                return createErrorResponse("0", -32602, "Invalid server key");
            }
        }
        
        // 解析消息为JsonRpcRequest
        JsonRpcRequest request = objectMapper.readValue(message, JsonRpcRequest.class);
        
        // 获取或创建客户端实例
        McpClientService client = getOrCreateClient(serverKey, serverConfig);
        if (client == null) {
            log.warn("Failed to get or create client for server: {}", serverKey);
            return createErrorResponse(request.getId().toString(), -32603, "Failed to connect to MCP server");
        }
        
        // 设置通知回调
        if (proxyInfo != null && proxyInfo.getSessionId() != null) {
            setupNotificationCallback(client, proxyInfo.getSessionId());
        }
        
        // 转发请求
        JsonRpcResponse response = client.forwardMessage(request);
        
        // 更新代理信息
        if (proxyInfo != null) {
            if (response != null) {
                proxyInfo.setResponseResult(response);
                
                // 初始化请求特殊处理
                if (JsonRpcConstants.Method.INITIALIZE.equals(request.getMethod()) && response.getResult() != null) {
                    try {
                        JsonNode resultNode = response.getResult();
                        String serverName = resultNode.path("server_info").path("name").asText("");
                        String serverVersion = resultNode.path("server_info").path("version").asText("");
                        
                        proxyInfo.setServerName(serverName);
                        proxyInfo.setServerVersion(serverVersion);
                        
                        // 记录服务器信息，但不修改ServerConfigDto，因为它没有这些字段
                        log.info("Server info from initialize response - name: {}, version: {}", 
                            serverName, serverVersion);
                    } catch (Exception e) {
                        log.error("Error extracting server info from initialize response: {}", e.getMessage());
                    }
                }
            } else {
                proxyInfo.setResponseError("No response from MCP server");
            }
            
            // 存储或更新代理信息
            proxyInfoService.storeProxyInfo(proxyInfo.getSessionId(), proxyInfo);
        }
        
        // 处理响应
        if (response == null) {
            return "{}"; // 空响应，通常是通知消息
        }
        
        return objectMapper.writeValueAsString(response);
    }

    @Override
    public JsonRpcResponse forwardMcpRequest(String serverKey, JsonRpcRequest jsonRpcRequest, ProxyInfoDto proxyInfo) {
        log.debug("Forwarding MCP request for server key {}: {}", serverKey, jsonRpcRequest);
        try {
            // 获取服务器配置
            ServerConfigDto serverConfig;
            if (proxyInfo != null && proxyInfo.getMcpServerId() != null && !proxyInfo.getMcpServerId().isEmpty()) {
                serverConfig = mcpServerService.getServerConfig(proxyInfo.getMcpServerId());
                if (serverConfig == null) {
                    log.warn("Invalid MCP server ID: {}", proxyInfo.getMcpServerId());
                    return JsonRpcResponse.error(jsonRpcRequest != null ? jsonRpcRequest.getId() : null, 
                           JsonRpcError.invalidParams("Invalid MCP server ID"));
                }
            } else {
                serverConfig = mcpServerService.getServerConfig(serverKey);
                if (serverConfig == null) {
                    log.warn("Invalid server key: {}", serverKey);
                    return JsonRpcResponse.error(jsonRpcRequest != null ? jsonRpcRequest.getId() : null, 
                           JsonRpcError.invalidParams("Invalid server key"));
                }
            }
            
            // 获取或创建客户端实例
            McpClientService client = getOrCreateClient(serverKey, serverConfig);
            if (client == null) {
                log.warn("Failed to get or create client for server: {}", serverKey);
                return JsonRpcResponse.error(jsonRpcRequest != null ? jsonRpcRequest.getId() : null, 
                       JsonRpcError.proxyError("Failed to connect to MCP server"));
            }
            
            // 设置通知回调
            if (proxyInfo != null && proxyInfo.getSessionId() != null) {
                setupNotificationCallback(client, proxyInfo.getSessionId());
            }
            
            // 转发请求
            JsonRpcResponse response = client.forwardMessage(jsonRpcRequest);
            
            // 更新代理信息
            if (proxyInfo != null && response != null) {
                proxyInfo.setResponseResult(response);
                proxyInfo.setResponseTime(LocalDateTime.now());
                
                if (response.getError() != null) {
                    proxyInfo.setResponseError(response.getError().getMessage());
                }
                
                if (proxyInfo.getRequestTime() != null) {
                    proxyInfo.setCostTime(
                        proxyInfo.getResponseTime().compareTo(proxyInfo.getRequestTime()) <= 0 ? 0L :
                        proxyInfo.getResponseTime().toEpochSecond(java.time.ZoneOffset.UTC) - 
                        proxyInfo.getRequestTime().toEpochSecond(java.time.ZoneOffset.UTC)
                    );
                }
                
                // 保存工具调用日志 - 完整支持tools/call和tools/list
                if (jsonRpcRequest != null) {
                    String method = jsonRpcRequest.getMethod();
                    if (appConfig.getSaveLog() && 
                        (method != null && (method.equals("tools/call") || method.equals("tools/list")))) {
                        logToolCall(proxyInfo);
                    }
                }
                
                // 保存更新后的代理信息
                if (proxyInfo.getSessionId() != null) {
                    storeProxyInfo(proxyInfo.getSessionId(), proxyInfo);
                }
            }
            
            return response;
        } catch (Exception e) {
            log.error("Error forwarding MCP request: {}", e.getMessage(), e);
            return JsonRpcResponse.error(
                jsonRpcRequest != null ? jsonRpcRequest.getId() : null,
                JsonRpcError.proxyError("Error forwarding request: " + e.getMessage())
            );
        }
    }
    
    /**
     * 获取或创建客户端实例
     */
    protected McpClientService getOrCreateClient(String serverKey, ServerConfigDto serverConfig) {
        McpClientService client = clientCache.get(serverKey);
        
        // 检查客户端是否存在且仍然连接
        if (client != null && client.isConnected()) {
            log.debug("Using existing client for server: {}", serverKey);
            return client;
        }
        
        // 如果客户端不存在或未连接，则创建新客户端
        try {
            client = new McpClientServiceImpl(objectMapper, null);
            boolean connected = client.connect(serverConfig);
            
            if (connected) {
                // 存储到缓存
                clientCache.put(serverKey, client);
                log.debug("Created new client for server: {}", serverKey);
                
                // 设置进程健康监视器
                if (serverConfig.getCommand() != null && !serverConfig.getCommand().isEmpty()) {
                    scheduleProcessHealthCheck(serverKey, client);
                }
                
                return client;
            } else {
                log.warn("Failed to connect to MCP server: {}", client.getError());
                return null;
            }
        } catch (Exception e) {
            log.error("Error creating MCP client: {}", e.getMessage());
            return null;
        }
    }
    
    /**
     * 调度进程健康检查任务
     * 定期检查进程健康状态，如果进程已终止则清理缓存
     */
    private void scheduleProcessHealthCheck(String serverKey, McpClientService client) {
        taskExecutor.scheduleAtFixedRate(() -> {
            try {
                if (!client.isConnected()) {
                    log.warn("Process for server {} is no longer connected, removing from cache", serverKey);
                    clientCache.remove(serverKey);
                }
            } catch (Exception e) {
                log.error("Error checking process health: {}", e.getMessage());
            }
        }, 60, 60, TimeUnit.SECONDS);
    }
    
    /**
     * 设置通知回调
     */
    protected void setupNotificationCallback(McpClientService client, String sessionId) {
        if (client != null) {
            client.setNotificationCallback(notification -> {
                try {
                    log.debug("Received notification from server: {}", notification);
                    // 处理通知，例如通过SSE发送给客户端
                    String notificationJson = objectMapper.writeValueAsString(notification);
                    sendSseEvent(sessionId, "notification", notificationJson);
                } catch (Exception e) {
                    log.error("Error processing notification: {}", e.getMessage());
                }
            });
        }
    }
    
    /**
     * 生成错误响应JSON字符串
     */
    protected String createErrorResponse(String id, int code, String message) throws Exception {
        try {
            JsonRpcResponse response = new JsonRpcResponse();
            response.setId(id);
            response.setJsonrpc("2.0");
            
            JsonRpcError error = new JsonRpcError(code, message, null);
            response.setError(error);
            
            return objectMapper.writeValueAsString(response);
        } catch (Exception e) {
            log.error("Error creating error response: {}", e.getMessage());
            return "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Internal error\"},\"id\":null}";
        }
    }

    @Override
    @Deprecated
    public void saveProxyLog(ProxyInfoDto proxyInfo) {
        logToolCall(proxyInfo);
    }

    @Override
    public void logToolCall(ProxyInfoDto proxyInfo) {
        if (proxyInfo == null || !appConfig.getSaveLog()) {
            return;
        }
        
        try {
            // 转换为ServerLog对象
            ServerLog serverLog = new ServerLog();
            
            // 基本信息
            serverLog.setSessionId(proxyInfo.getSessionId());
            serverLog.setRequestId(proxyInfo.getRequestId() != null ? proxyInfo.getRequestId().toString() : null);
            serverLog.setRequestFrom(proxyInfo.getRequestFrom());
            serverLog.setServerUuid(proxyInfo.getServerUuid());
            serverLog.setServerKey(proxyInfo.getServerKey());
            serverLog.setServerName(proxyInfo.getServerName());
            serverLog.setServerVersion(proxyInfo.getServerVersion());
            serverLog.setMcpServerId(proxyInfo.getMcpServerId());
            
            // 客户端信息
            serverLog.setClientName(proxyInfo.getClientName());
            serverLog.setClientVersion(proxyInfo.getClientVersion());
            serverLog.setProtocolVersion(proxyInfo.getProtocolVersion());
            serverLog.setJsonrpcVersion(proxyInfo.getJsonrpcVersion());
            
            // 请求信息
            serverLog.setRequestMethod(proxyInfo.getRequestMethod());
            String params = proxyInfo.getRequestParams() != null ? 
                            objectMapper.writeValueAsString(proxyInfo.getRequestParams()) : null;
            serverLog.setRequestParams(params);
            
            // 提取工具名称并添加到参数中（如果是工具调用）
            String toolName = "unknown tool";
            if ("tools/call".equals(proxyInfo.getRequestMethod()) && proxyInfo.getRequestParams() != null) {
                try {
                    Map<String, Object> paramsMap = objectMapper.convertValue(proxyInfo.getRequestParams(), Map.class);
                    if (paramsMap.containsKey("name")) {
                        toolName = (String) paramsMap.get("name");
                        
                        // 将工具名称添加到ServerLog的适当字段
                        serverLog.setServerConfigName(toolName); // 使用serverConfigName字段存储工具名称
                    }
                } catch (Exception e) {
                    log.warn("Failed to extract tool name from params: {}", e.getMessage());
                }
            }
            
            // 响应信息
            String result = proxyInfo.getResponseResult() != null ? 
                            objectMapper.writeValueAsString(proxyInfo.getResponseResult()) : null;
            serverLog.setResponseResult(result);
            
            // 错误信息
            serverLog.setResponseError(proxyInfo.getResponseError() != null ? 
                                      proxyInfo.getResponseError().toString() : null);
            
            // 时间信息
            serverLog.setConnectionTime(proxyInfo.getConnectionTime());
            serverLog.setRequestTime(proxyInfo.getRequestTime());
            serverLog.setResponseTime(proxyInfo.getResponseTime());
            serverLog.setCostTime(proxyInfo.getCostTime());
            
            // 创建和更新时间
            LocalDateTime now = LocalDateTime.now();
            serverLog.setCreatedAt(now);
            serverLog.setUpdatedAt(now);
            
            // 保存日志
            serverLogMapper.insert(serverLog);
            
            log.debug("Tool call logged: {} - {}", proxyInfo.getRequestMethod(), toolName);
        } catch (Exception e) {
            log.error("Failed to log tool call: {}", e.getMessage(), e);
        }
    }

    /**
     * 为会话创建消息队列
     *
     * @param sessionId 会话ID
     */
    private void createMessageQueue(String sessionId) {
        if (sessionId == null || sessionId.isEmpty() || messageQueues.containsKey(sessionId)) {
            return;
        }
        
        // 创建消息队列
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE);
        messageQueues.put(sessionId, queue);
        
        // 创建消息处理器
        ScheduledExecutorService processor = Executors.newSingleThreadScheduledExecutor();
        messageProcessors.put(sessionId, processor);
        
        // 启动消息处理任务
        processor.scheduleWithFixedDelay(() -> processMessageQueue(sessionId), 0, 100, TimeUnit.MILLISECONDS);
        
        log.debug("Created message queue for session {}", sessionId);
    }
    
    /**
     * 处理消息队列
     *
     * @param sessionId 会话ID
     */
    private void processMessageQueue(String sessionId) {
        BlockingQueue<String> queue = messageQueues.get(sessionId);
        if (queue == null || queue.isEmpty()) {
            return;
        }
        
        try {
            // 从队列中取出消息
            String message = queue.poll();
            if (message == null) {
                return;
            }
            
            // 获取代理信息
            ProxyInfoDto proxyInfo = getProxyInfo(sessionId);
            if (proxyInfo == null) {
                log.warn("No proxy info found for session {}, cannot process message", sessionId);
                return;
            }
            
            // 处理消息，包含重试逻辑
            processMessageWithRetry(proxyInfo.getServerKey(), message, proxyInfo, sessionId, 0);
        } catch (Exception e) {
            log.error("Error processing message queue for session {}: {}", sessionId, e.getMessage(), e);
        }
    }
    
    /**
     * 处理消息，包含重试逻辑
     *
     * @param serverKey 服务器键
     * @param message 消息内容
     * @param proxyInfo 代理信息
     * @param sessionId 会话ID
     * @param retryCount 当前重试次数
     */
    private void processMessageWithRetry(String serverKey, String message, ProxyInfoDto proxyInfo, 
                                       String sessionId, int retryCount) {
        try {
            // 处理消息
            String response = doProcessSseMessage(serverKey, message, proxyInfo);
            
            // 发送响应
            if (response != null) {
                sendSseEvent(sessionId, "message", response);
            }
        } catch (Exception e) {
            log.error("Failed to process message for session {}, retry {}: {}", 
                      sessionId, retryCount, e.getMessage(), e);
            
            // 如果未达到最大重试次数，则重试
            if (retryCount < MAX_RETRY_COUNT) {
                ScheduledExecutorService processor = messageProcessors.get(sessionId);
                if (processor != null && !processor.isShutdown()) {
                    processor.schedule(() -> 
                        processMessageWithRetry(serverKey, message, proxyInfo, sessionId, retryCount + 1),
                        RETRY_DELAY_MS, TimeUnit.MILLISECONDS);
                }
            } else {
                log.error("Max retry count reached for message processing, session: {}", sessionId);
                try {
                    // 发送错误响应
                    String errorResponse = createErrorResponse("0", -32603, 
                            "Failed to process message after " + MAX_RETRY_COUNT + " retries");
                    sendSseEvent(sessionId, "message", errorResponse);
                } catch (Exception ex) {
                    log.error("Failed to send error response: {}", ex.getMessage(), ex);
                }
            }
        }
    }
    
    /**
     * 停止消息队列
     *
     * @param sessionId 会话ID
     */
    private void stopMessageQueue(String sessionId) {
        // 移除消息队列
        messageQueues.remove(sessionId);
        
        // 停止并移除消息处理器
        ScheduledExecutorService processor = messageProcessors.remove(sessionId);
        if (processor != null) {
            processor.shutdown();
            try {
                if (!processor.awaitTermination(5, TimeUnit.SECONDS)) {
                    processor.shutdownNow();
                }
            } catch (InterruptedException e) {
                processor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        
        log.debug("Stopped message queue for session {}", sessionId);
    }

    /**
     * 实现接口方法: 注册SSE发射器
     */
    @Override
    public void registerSseEmitter(String sessionId, SseEmitter emitter) {
        if (sessionId == null || sessionId.isEmpty() || emitter == null) {
            log.warn("Invalid session ID or emitter for registration");
            return;
        }
        
        log.debug("Registering SSE emitter for session: {}", sessionId);
        
        // 存储SSE发射器
        sseEmitters.put(sessionId, emitter);
        
        // 设置超时处理
        emitter.onTimeout(() -> {
            log.debug("SSE connection timed out for session: {}", sessionId);
            unregisterSseEmitter(sessionId);
        });
        
        // 设置完成处理
        emitter.onCompletion(() -> {
            log.debug("SSE connection completed for session: {}", sessionId);
            unregisterSseEmitter(sessionId);
        });
        
        // 设置错误处理
        emitter.onError(ex -> {
            log.warn("SSE connection error for session {}: {}", sessionId, ex.getMessage());
            unregisterSseEmitter(sessionId);
        });
        
        // 启动心跳调度器
        startHeartbeatScheduler(sessionId);
        
        // 发送初始连接成功事件
        try {
            sendSseEvent(sessionId, "connected", "{\"status\":\"connected\",\"sessionId\":\"" + sessionId + "\"}");
        } catch (Exception e) {
            log.error("Failed to send initial connection event: {}", e.getMessage());
        }
    }

    /**
     * 实现接口方法: 注销SSE发射器
     */
    @Override
    public void unregisterSseEmitter(String sessionId) {
        if (sessionId == null || sessionId.isEmpty()) {
            return;
        }
        
        log.debug("Unregistering SSE emitter for session: {}", sessionId);
        
        // 获取并移除SSE发射器
        SseEmitter emitter = sseEmitters.remove(sessionId);
        if (emitter != null) {
            try {
                emitter.complete();
            } catch (Exception e) {
                log.warn("Error completing SSE emitter: {}", e.getMessage());
            }
        }
        
        // 停止心跳调度器
        stopHeartbeatScheduler(sessionId);
        
        // 停止消息队列
        stopMessageQueue(sessionId);
    }

    /**
     * 实现接口方法: 发送SSE事件
     */
    @Override
    public void sendSseEvent(String sessionId, String eventName, String eventData) {
        if (sessionId == null || sessionId.isEmpty()) {
            log.warn("Invalid session ID for sending SSE event");
            return;
        }
        
        SseEmitter emitter = sseEmitters.get(sessionId);
        if (emitter == null) {
            log.warn("No SSE emitter found for session: {}", sessionId);
            return;
        }
        
        try {
            emitter.send(SseEmitter.event()
                    .name(eventName)
                    .data(eventData, MediaType.APPLICATION_JSON));
            log.debug("Sent SSE event '{}' to session {}", eventName, sessionId);
        } catch (IOException e) {
            log.error("Failed to send SSE event to session {}: {}", sessionId, e.getMessage());
            unregisterSseEmitter(sessionId);
        }
    }

    /**
     * 实现接口方法: 发送心跳
     */
    @Override
    public boolean sendHeartbeat(String sessionId) {
        if (sessionId == null || sessionId.isEmpty()) {
            return false;
        }
        
        SseEmitter emitter = sseEmitters.get(sessionId);
        if (emitter == null) {
            log.warn("No SSE emitter found for heartbeat, session: {}", sessionId);
            return false;
        }
        
        try {
            // 发送心跳事件
            emitter.send(SseEmitter.event()
                    .name("heartbeat")
                    .data("{\"time\":" + System.currentTimeMillis() + "}", MediaType.APPLICATION_JSON));
            log.debug("Sent heartbeat to session {}", sessionId);
            return true;
        } catch (IOException e) {
            log.error("Failed to send heartbeat to session {}: {}", e.getMessage());
            unregisterSseEmitter(sessionId);
            return false;
        }
    }

    /**
     * 启动心跳调度器
     */
    private void startHeartbeatScheduler(String sessionId) {
        if (sessionId == null || sessionId.isEmpty() || heartbeatSchedulers.containsKey(sessionId)) {
            return;
        }
        
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            try {
                sendHeartbeat(sessionId);
            } catch (Exception e) {
                log.error("Error sending heartbeat: {}", e.getMessage());
            }
        }, 0, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
        
        heartbeatSchedulers.put(sessionId, scheduler);
        log.debug("Started heartbeat scheduler for session {}", sessionId);
    }

    /**
     * 停止心跳调度器
     */
    private void stopHeartbeatScheduler(String sessionId) {
        if (sessionId == null || sessionId.isEmpty()) {
            return;
        }
        
        ScheduledExecutorService scheduler = heartbeatSchedulers.remove(sessionId);
        if (scheduler != null) {
            scheduler.shutdown();
            try {
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
            log.debug("Stopped heartbeat scheduler for session {}", sessionId);
        }
    }
}